package com.li.filter;

import org.apache.rocketmq.client.hook.ConsumeMessageContext;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.MDC;

import java.util.Iterator;

/**
 * @program: rocketmq_boot
 * @description:
 * @author: li
 * @create: 2023-03-08 10:37
 **/
public class RocketMqConsumerMessageHook implements ConsumeMessageHook {
    @Override
    public String hookName() {
        return "RocketMqConsumerMessageHook";
    }

    @Override
    public void consumeMessageBefore(ConsumeMessageContext context) {
        Iterator<MessageExt> iterator = context.getMsgList().iterator();
        while (iterator.hasNext()) {
            MessageExt messageExt = iterator.next();
            String traceId = messageExt.getUserProperty("traceId");
            // 从消息中拿出traceId 放入当前线程的MDC
            MDC.put("traceId", traceId);
            System.out.println("traceId-" + traceId);
        }
    }

    @Override
    public void consumeMessageAfter(ConsumeMessageContext context) {
    }
}
